package com.example.kafkademo.consumer;

import io.vertx.core.Vertx;
import io.vertx.core.json.JsonObject;
import io.vertx.kafka.client.consumer.KafkaConsumer;
import io.vertx.kafka.client.serialization.JsonObjectDeserializer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Arrays;
import java.util.Properties;

public class VertxConsumer {

    public static void main(String[] args) {

        Vertx vertx=Vertx.vertx();
        final Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.1.14:9092");  //kafka主机地址

        props.put("group.id", "GroupA");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");

        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", JsonObjectDeserializer.class.getName());


        KafkaConsumer<String, JsonObject> consumer = KafkaConsumer.create(vertx, props);
        consumer.subscribe("huahua");

        while (true) {
            //kafka消费者读取有问题,在控制台显示是中文乱码
            consumer.poll(1000, ar ->{
                for (ConsumerRecord<String, JsonObject> record : ar.result().records()) {
                    System.out.printf("offset = %d, key = %s, value = %s", record.offset(), record.key(), record.value() + "\n");
                }
                if(ar.result().isEmpty()){
                    System.out.println("消息为空");
                }
            });

        }
    }

}
